package com.example.controller;

import com.example.mq.BizMQPushConsumer;
import com.example.mq.GrayMQPushConsumer;
import com.example.mq.GrayMQPushConsumerFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@Slf4j
@RestController
@RequestMapping("/api/mq")
public class MqController {

    private static DefaultMQProducer producer;
    
    static {
        producer = new DefaultMQProducer("producerGroup");
        producer.setNamesrvAddr("127.0.0.1:9876");
        try {
            producer.start();
        } catch (MQClientException e) {
            log.error(e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }

    @GetMapping("/consumerStart")
    public ResponseEntity<String> consumerStart() {
        try {
            DefaultMQPushConsumer consumer = new GrayMQPushConsumer();
            consumer.setConsumerGroup("consumerGroup1");
            consumer.setNamesrvAddr("127.0.0.1:9876");
            consumer.subscribe("testTopic", "*");
            consumer.registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
                log.info("consumerGroup1, consumer1, msgId={}", messages.get(0).getMsgId());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();

            consumer = new BizMQPushConsumer();
            consumer.setConsumerGroup("consumerGroup1");
            consumer.setNamesrvAddr("127.0.0.1:9876");
            consumer.subscribe("testTopic", "*");
            consumer.registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
                log.info("consumerGroup1, consumer2, msgId={}", messages.get(0).getMsgId());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();

            consumer = new GrayMQPushConsumer();
            consumer.setConsumerGroup("consumerGroup2");
            consumer.setNamesrvAddr("127.0.0.1:9876");
            consumer.subscribe("testTopic", "*");
            consumer.registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
                log.info("consumerGroup2, consumer1, msgId={}", messages.get(0).getMsgId());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();

            consumer = new BizMQPushConsumer();
            consumer.setConsumerGroup("consumerGroup2");
            consumer.setNamesrvAddr("127.0.0.1:9876");
            consumer.subscribe("testTopic", "*");
            consumer.registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
                log.info("consumerGroup2, consumer2, msgId={}", messages.get(0).getMsgId());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
            log.info("consumerStarted");
            return ResponseEntity.ok("");
        } catch (Exception e) {
            log.error("Error in mq test", e);
            return ResponseEntity.internalServerError().body("Error: " + e.getMessage());
        }
    }

    @GetMapping("/msgSend")
    public ResponseEntity<String> msgSend() {
        try {
            Message msg = new Message();
            msg.setTopic("testTopic");
            msg.setBody("hello, world".getBytes());
            producer.send(msg);
            log.info("msgSended");
            return ResponseEntity.ok("");
        } catch (Exception e) {
            log.error("Error in mq test", e);
            return ResponseEntity.internalServerError().body("Error: " + e.getMessage());
        }
    }

    @GetMapping("/consumersShutdown")
    public ResponseEntity<String> consumersShutdown() {
        try {
            GrayMQPushConsumerFactory.shutDownAll();
            log.info("consumersShutdowned");
            return ResponseEntity.ok("");
        } catch (Exception e) {
            log.error("Error in mq test", e);
            return ResponseEntity.internalServerError().body("Error: " + e.getMessage());
        }
    }
}